package com.xzc.apitest.tabletest

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, OldCsv, Schema}
import org.apache.flink.types.Row

object FileOutputTest {
  def main(args: Array[String]): Unit = {
    //1.创建环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val tableEnv = StreamTableEnvironment.create(env)

    //2.1 读取文件
    val filePath = "D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE()))
      .createTemporaryTable("inputTable")

    //3 转换操作
    val sensorTable = tableEnv.from("inputTable")
    //3.1 简单转换
    val resultTable = sensorTable
      .select('id, 'temp)
      .filter('id === "sensor_1")
    //3.2 聚合转换
    val aggTable = sensorTable
      .groupBy('id)
      .select('id, 'id.count as 'count)

    //4 输出文件
    //注册输出表
    val outputfilePath = "D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\output.txt"
    tableEnv.connect(new FileSystem().path(outputfilePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("temp", DataTypes.DOUBLE()))
      .createTemporaryTable("outputTable")
    resultTable.insertInto("outputTable")
    tableEnv.connect(new FileSystem().path(outputfilePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("cnt", DataTypes.BIGINT()))
      .createTemporaryTable("aggTable")
    //AppendStreamTableSink requires that Table has only insert changes
    //因为输出是文件，是不可改的，而aggTable的输出是要可以改的，所以这里不能输出到文件
    aggTable.insertInto("aggTable")

    resultTable.toAppendStream[(String, Double)].print("result")
    //聚合的结果不能append
    //false代表之前的对应的数据已失效，后面true才是新的
    //如TIM群中撤回了一条消息
    aggTable.toRetractStream[(String, Long)].print("agg")
    aggTable.toRetractStream[Row].print("agg")
    //upsert模式无法在流中实现，因为这个操作取决于外部系统


    env.execute("file output test")
  }

}
